package org.tao.lightningmq.client.producer;

import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.client.common.CommonCache;
import org.tao.lightningmq.client.netty.BrokerRemoteRespHandler;
import org.tao.lightningmq.common.dto.*;
import org.tao.lightningmq.common.enums.*;
import org.tao.lightningmq.common.event.EventBus;
import org.tao.lightningmq.common.remote.BrokerNettyRemoteClient;
import org.tao.lightningmq.common.remote.NameServerNettyRemoteClient;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.common.transaction.TransactionListener;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author lht
 * @date 2025/1/19 - 17:09
 * @description:
 */
public class DefaultProducer implements Producer {

    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProducer.class);
    /**
     * nameserver集群的地址，用;分隔
     * 如：127.0.0.1:28080;127.0.0.1:28081
     */
    private String nameserverAddrStr;
    private String nameServerIp;
    private int nameServerPort;
    private String nameServerUser;
    private String nameServerPassword;
    private String brokerClusterGroup;
    private String brokerRole = "single";
    private List<String> brokerAddressList; //broker地址会有多个？broker节点会有多个，水平扩展的效果，水平扩展（存储内容会增加，承载压力也会大大增加，节点的选择问题）
    private List<String> masterAddressList;
    private String producerId;
    private NameServerNettyRemoteClient nameServerNettyRemoteClient;
    private TransactionListener transactionListener;
    private Map<String, BrokerNettyRemoteClient> brokerNettyRemoteClientMap = new ConcurrentHashMap<>();
    private final Map<String, NameServerNettyRemoteClient> nameServerNettyRemoteClientMap = new ConcurrentHashMap<>();

    @Override
    public void start() {
        boolean isRegistrySuccess = false;
        for (String nameserverAddr : nameserverAddrStr.split(";")) {
            String[] split = nameserverAddr.split(":");
            NameServerNettyRemoteClient ns = new NameServerNettyRemoteClient(split[0], Integer.parseInt(split[1]));
            ns.buildConnection();
            ns.setNameServerLogoutCallBack(() -> {
                ns.buildConnection();
                doRegistry(ns);
            });
            isRegistrySuccess |= doRegistry(ns);
            LOGGER.info("producer registry to {} result is: {}", nameserverAddr, isRegistrySuccess);
            nameServerNettyRemoteClientMap.put(nameserverAddr, ns);
        }
//        nameServerNettyRemoteClient = new NameServerNettyRemoteClient(nameServerIp, nameServerPort);
//        nameServerNettyRemoteClient.buildConnection();
//        boolean isRegistrySuccess = this.doRegistry();

        this.setProducerId(UUID.randomUUID().toString());
        if (this.transactionListener != null) {
            CommonCache.getTransactionListenerMap().put(this.getProducerId(), this.transactionListener);
        }
        if (isRegistrySuccess) {
            this.startHeartBeatTask();
            this.fetchBrokerAddress();
            this.connectBroker();
            this.startRefreshBrokerAddressJob();
        }
    }

    @Override
    public SendResult send(MessageDTO messageDTO) {
        BrokerNettyRemoteClient remoteClient = this.getBrokerNettyRemoteClient();
        String msgId = UUID.randomUUID().toString();
        messageDTO.setMsgId(msgId);
        messageDTO.setSendWay(MessageSendWay.SYNC.getCode());
        TcpMsg tcpMsg = new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), JSON.toJSONBytes(messageDTO));
        TcpMsg respMsg = remoteClient.sendSyncMsg(tcpMsg, msgId);
        SendMessageToBrokerResponseDTO sendMessageToBrokerResponseDTO = JSON.parseObject(respMsg.getBody(), SendMessageToBrokerResponseDTO.class);
        int responseStatus = sendMessageToBrokerResponseDTO.getStatus();
        if (responseStatus == SendMessageToBrokerResponseStatus.SUCCESS.getCode()) {
            return SendResult.success();
        }
        LOGGER.warn("send mq fail:{}", sendMessageToBrokerResponseDTO.getDesc());
        return SendResult.fail();
    }

    @Override
    public void sendAsync(MessageDTO messageDTO) {
        BrokerNettyRemoteClient remoteClient = this.getBrokerNettyRemoteClient();
        messageDTO.setSendWay(MessageSendWay.ASYNC.getCode());
        TcpMsg tcpMsg = new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), JSON.toJSONBytes(messageDTO));
        remoteClient.sendAsyncMsg(tcpMsg);
    }

    private BrokerNettyRemoteClient getBrokerNettyRemoteClient() {
        return new ArrayList<>(this.getBrokerNettyRemoteClientMap().values()).get(0);
    }

    private boolean doRegistry(NameServerNettyRemoteClient nsClient) {
        String registryMsgId = UUID.randomUUID().toString();
        ServiceRegistryReqDTO serviceRegistryReqDTO = new ServiceRegistryReqDTO();
        serviceRegistryReqDTO.setMsgId(registryMsgId);
        serviceRegistryReqDTO.setUser(nameServerUser);
        serviceRegistryReqDTO.setPassword(nameServerPassword);
        serviceRegistryReqDTO.setRegistryType(RegistryTypeEnum.PRODUCER.getCode());
        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.REGISTRY.getCode(), JSON.toJSONBytes(serviceRegistryReqDTO));
        TcpMsg registryResponse = nsClient.sendSyncMsg(tcpMsg, registryMsgId);
        if (NameServerResponseCode.REGISTRY_SUCCESS.getCode() == registryResponse.getCode()) {
            return true;
        } else {
            LOGGER.warn("注册账号失败");
            return false;
        }
    }

    private void startHeartBeatTask() {
        Thread heartBeatTask = new Thread(() -> {
            LOGGER.info("Producer心跳数据发送任务启动");
            HeartBeatDTO heartBeatDTO = new HeartBeatDTO();
            while (true) {
                try {
                    TimeUnit.SECONDS.sleep(3);
                    LOGGER.info("Producer执行心跳数据发送");
                    String heartBeatMsgId = UUID.randomUUID().toString();
                    heartBeatDTO.setMsgId(heartBeatMsgId);
                    TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.HEART_BEAT.getCode(), JSON.toJSONBytes(heartBeatDTO));
                    for (NameServerNettyRemoteClient nsClient : nameServerNettyRemoteClientMap.values()) {
                        if (nsClient.isActive()) {
                            nsClient.sendAsyncMsg(tcpMsg);
                        }
                    }
//                    TcpMsg heartBeatResponse = nameServerNettyRemoteClient.sendSyncMsg(, heartBeatMsgId);
//                    LOGGER.info("Producer heart beat response data is :{}", JSON.parseObject(heartBeatResponse.getBody()));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "producer-heart-beat-task");
        heartBeatTask.start();
    }

    public void startRefreshBrokerAddressJob() {
        Thread refreshBrokerAddressJob = new Thread(() -> {
            while (true) {
                try {
                    TimeUnit.SECONDS.sleep(3);
                    fetchBrokerAddress();
                    connectBroker();
                } catch (Exception e) {
                    LOGGER.error("refresh broker address job error:", e);
                }
            }
        }, "producer-refresh-broker-address-job");
        refreshBrokerAddressJob.start();
    }

    private void fetchBrokerAddress() {
        NameServerNettyRemoteClient nsClient = getUsefulClient();
        String msgId = UUID.randomUUID().toString();
        PullBrokerIpDTO pullBrokerIpDTO = new PullBrokerIpDTO();
        if (this.brokerClusterGroup != null) {
            this.setBrokerRole("master");
            pullBrokerIpDTO.setBrokerClusterGroup(this.brokerClusterGroup);
        }
        pullBrokerIpDTO.setMsgId(msgId);
        pullBrokerIpDTO.setRole(this.brokerRole);
        TcpMsg tcpMsg = nsClient.sendSyncMsg(new TcpMsg(
                NameServerEventCode.PULL_BROKER_IP_LIST.getCode(),
                JSON.toJSONBytes(pullBrokerIpDTO)), msgId);
        PullBrokerIpRespDTO pullBrokerIpRespDTO = JSON.parseObject(tcpMsg.getBody(), PullBrokerIpRespDTO.class);
        // broker地址, slave是只读的所以没必要
        this.brokerAddressList = pullBrokerIpRespDTO.getAddressList();
        this.masterAddressList = pullBrokerIpRespDTO.getMasterAddressList();
        LOGGER.info("fetch broker address:{}, master:{}", this.getBrokerAddressList(), this.getMasterAddressList());

    }

    private NameServerNettyRemoteClient getUsefulClient() {
        for (NameServerNettyRemoteClient value : nameServerNettyRemoteClientMap.values()) {
            if (value.isActive()) {
                return value;
            }
        }
        throw new RuntimeException("没有已连接的Nameserver节点");
    }

    private void connectBroker() {
        List<String> brokerAddressList = Collections.emptyList();
        if ("master".equals(this.brokerRole)) {
            if (this.masterAddressList == null || this.masterAddressList.isEmpty()) {
                throw new RuntimeException("broker地址不能为空");
            }
            brokerAddressList = this.masterAddressList;
        } else if ("single".equals(this.brokerRole)) {
            if (this.brokerAddressList == null || this.brokerAddressList.isEmpty()) {
                throw new RuntimeException("broker地址不能为空");
            }
            brokerAddressList = this.brokerAddressList;
        }
        List<BrokerNettyRemoteClient> newBrokerNettyRemoteClientList = brokerAddressList.stream()
                .map(brokerAddr -> {
                    BrokerNettyRemoteClient nettyRemoteClient = this.brokerNettyRemoteClientMap.get(brokerAddr);
                    if (nettyRemoteClient != null && nettyRemoteClient.isChannelActive()) {
                        return nettyRemoteClient;
                    }
                    // 对于未连接的broker，建立连接
                    String[] split = brokerAddr.split(":");
                    String ip = split[0], port = split[1];
                    BrokerNettyRemoteClient brokerNettyRemoteClient = new BrokerNettyRemoteClient(ip, Integer.parseInt(port));
                    brokerNettyRemoteClient.buildConnection(new BrokerRemoteRespHandler(new EventBus("consumer-client-eventbus")));
                    return brokerNettyRemoteClient;
                }).collect(Collectors.toList());
        List<String> finalBrokerAddressList = brokerAddressList;
        this.brokerNettyRemoteClientMap.keySet().stream()
                .filter(brokerAddr -> !finalBrokerAddressList.contains(brokerAddr))
                .forEach(brokerAddr -> {
                    // 关闭不需要的连接
                    BrokerNettyRemoteClient brokerNettyRemoteClient = this.brokerNettyRemoteClientMap.get(brokerAddr);
                    brokerNettyRemoteClient.close();
                });
        this.brokerNettyRemoteClientMap = newBrokerNettyRemoteClientList.stream()
                .collect(Collectors.toMap(BrokerNettyRemoteClient::getBrokerReqId, brokerNettyRemoteClient -> brokerNettyRemoteClient));
    }

    @Override
    public SendResult sendTxMessage(MessageDTO messageDTO) {
        Objects.requireNonNull(transactionListener, "transactionListener is null");
        String msgId = UUID.randomUUID().toString();
        messageDTO.setTxFlag(TxMessageFlagEnum.HALF_MSG.getCode());
        messageDTO.setMsgId(msgId);
        messageDTO.setProducerId(this.getProducerId());
        messageDTO.setSendWay(MessageSendWay.SYNC.getCode());
        BrokerNettyRemoteClient remoteClient = this.getBrokerNettyRemoteClient();
        TcpMsg tcpMsg = new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), JSON.toJSONBytes(messageDTO));
        TcpMsg responseMsg = remoteClient.sendSyncMsg(tcpMsg, msgId);
        boolean isHalfMsgSendSuccess = (responseMsg != null && responseMsg.getCode() == BrokerResponseCode.HALF_MSG_SEND_SUCCESS.getCode());
        if (!isHalfMsgSendSuccess) {
            throw new RuntimeException("half msg send fail");
        }
        LocalTransactionState localTransactionState = transactionListener.executeLocalTransaction(messageDTO);
        messageDTO.setLocalTxState(localTransactionState.getCode());
        if (LocalTransactionState.COMMIT.equals(localTransactionState)) {
            messageDTO.setTxFlag(TxMessageFlagEnum.REMAIN_HALF_ACK.getCode());
            TcpMsg remainHalfAckMsg = new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), JSON.toJSONBytes(messageDTO));
            TcpMsg remainHalfAckResp = remoteClient.sendSyncMsg(remainHalfAckMsg, msgId);
            LOGGER.info("sendTxMessage remainHalfAckResp :{}", remainHalfAckResp);
        } else if (LocalTransactionState.ROLLBACK.equals(localTransactionState)) {
            //应该要通知到broker本地事务消息执行失败这个行为，不然的话broker会一直回调客户端查询状态，这里会有额外的性能损耗
            messageDTO.setTxFlag(TxMessageFlagEnum.REMAIN_HALF_ACK.getCode());
            TcpMsg rollbackMsg = new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), com.alibaba.fastjson.JSON.toJSONBytes(messageDTO));
            TcpMsg rollbackAckResp = remoteClient.sendSyncMsg(rollbackMsg, msgId);
            LOGGER.info("sendTxMessage rollbackAckResp :{}", rollbackAckResp);
        } else if (LocalTransactionState.UNKNOWN.equals(localTransactionState)) {
            //等待broker回调查询进行状态判断
        }
        return SendResult.success();
    }

    public String getNameServerIp() {
        return nameServerIp;
    }

    public void setNameServerIp(String nameServerIp) {
        this.nameServerIp = nameServerIp;
    }

    public int getNameServerPort() {
        return nameServerPort;
    }

    public void setNameServerPort(int nameServerPort) {
        this.nameServerPort = nameServerPort;
    }

    public String getNameServerUser() {
        return nameServerUser;
    }

    public void setNameServerUser(String nameServerUser) {
        this.nameServerUser = nameServerUser;
    }

    public String getNameServerPassword() {
        return nameServerPassword;
    }

    public void setNameServerPassword(String nameServerPassword) {
        this.nameServerPassword = nameServerPassword;
    }

    public String getBrokerClusterGroup() {
        return brokerClusterGroup;
    }

    public void setBrokerClusterGroup(String brokerClusterGroup) {
        this.brokerClusterGroup = brokerClusterGroup;
    }

    public String getBrokerRole() {
        return brokerRole;
    }

    public void setBrokerRole(String brokerRole) {
        this.brokerRole = brokerRole;
    }

    public List<String> getBrokerAddressList() {
        return brokerAddressList;
    }

    public void setBrokerAddressList(List<String> brokerAddressList) {
        this.brokerAddressList = brokerAddressList;
    }

    public List<String> getMasterAddressList() {
        return masterAddressList;
    }

    public void setMasterAddressList(List<String> masterAddressList) {
        this.masterAddressList = masterAddressList;
    }

    public String getProducerId() {
        return producerId;
    }

    public void setProducerId(String producerId) {
        this.producerId = producerId;
    }

    public NameServerNettyRemoteClient getNameServerNettyRemoteClient() {
        return nameServerNettyRemoteClient;
    }

    public void setNameServerNettyRemoteClient(NameServerNettyRemoteClient nameServerNettyRemoteClient) {
        this.nameServerNettyRemoteClient = nameServerNettyRemoteClient;
    }

    public Map<String, BrokerNettyRemoteClient> getBrokerNettyRemoteClientMap() {
        return brokerNettyRemoteClientMap;
    }

    public void setTransactionListener(TransactionListener transactionListener) {
        this.transactionListener = transactionListener;
    }

    public String getNameserverAddrStr() {
        return nameserverAddrStr;
    }

    public void setNameserverAddrStr(String nameserverAddrStr) {
        this.nameserverAddrStr = nameserverAddrStr;
    }
}
